#!/usr/bin/env python3
# coding=utf-8

import sys
import importlib
importlib.reload(sys)
import pandas as pd
import getopt
import os
from datetime import timedelta, datetime
import datetime
import numpy
import subprocess
from fbprophet import Prophet

sys.path.append("/opt/mv-bash/mv-python")
from send_mail import send_mail

sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
from utils import cmdutils
import read_config


class Monitor:
    __TABLE_DATA_QUALITY = "dm_quality_metric_di"
    __CAL_TYPE_METRIC = 1
    __CAL_TYPE_DIM = 2
    __MONITOR_TYPE_HISTORY = 1

    def __init__(self, env, arg_date, monitor_table, monitor_db, warn_email, custom):
        self.__env = env
        self.__arg_date = arg_date
        self.__monitor_table = monitor_table
        self.__custom = custom
        if self.__env == 'test':
            self.__env = env
        else:
            self.__env = 'prod'
        if self.__env == 'test':
            self.__db = monitor_db
        else:
            self.__db = 'mediav_base'
        self.__warn_email = warn_email
        self.__table_data_quality = self.__db + "." + Monitor.__TABLE_DATA_QUALITY

    def do_execute(self):
        config = self.__custom
        row_all = 0
        result_all = []
        monitor_type_fin = 0
        cfg_fin = []
        for cfg in config:
            monitor_type = cfg.get('monitor_type')
            if monitor_type == 1:
                result, row = self.__do_monitor_metrics(cfg)
                row_all = row_all + row  # 用于dataframe的行号
                monitor_type_fin = monitor_type
                cfg_fin = cfg
                if result_all == []:
                    result_all = result
                else:
                    result_all = numpy.hstack((result_all, result))
            if monitor_type == 2:
                self.__do_monitor_mail(cfg)
            if monitor_type == 3:
                result, row = self.__do_monitor_metrics(cfg)
                self.__do_sendmail(result, row, cfg)

        if monitor_type_fin == 1:
            self.__do_sendmail(result_all, row_all, cfg_fin)

    def get_monitor_sqls(self, cfg):
        arg_date = self.__arg_date
        cur_date = datetime.datetime(int(arg_date.split("-")[0]), int(arg_date.split("-")[1]),
                                     int(arg_date.split("-")[2]))
        last_date = cur_date + datetime.timedelta(days=-1)
        table_data_quality = self.__table_data_quality
        env = self.__env
        db = self.__db
        table_name = self.__monitor_table
        monitor_type = cfg.get('monitor_type')
        sql_metas = []

        cols = cfg['cols']
        for col_name in cols:
            monitor_id = cfg.get('monitor_id')
            threshold_warning = cfg.get('monitor_algo').get('threshold').get('warning')
            threshold_error = cfg.get('monitor_algo').get('threshold').get('error')
            monitor_algo = cfg.get('monitor_algo').get('algo')
            monitor_days = cfg.get('monitor_algo').get('days')
            monitor_name = cfg.get('monitor_name')
            date_up_before = last_date.strftime("%Y-%m-%d")
            date_down_before = (cur_date + datetime.timedelta(days=float(-7*monitor_days))).strftime("%Y-%m-%d") % vars()
            sql_date_contitions = "date <= '%(date_up_before)s' and date >= '%(date_down_before)s'" % vars()

            cal_type = cfg.get('cal_type')
            dim_value_cfg = cfg.get('dim_value')
            if len(dim_value_cfg) == 1 and dim_value_cfg[0] == 'quality_metrics':  # top指标维度获取
                sql_value = '''select distinct dim_value
                                 from (select *
                                             ,row_number() over(partition by dim_value, rank order by etl_date desc) as num 
                                         from %(db)s.quality_metrics
                                        where monitor_id = %(monitor_id)s
                                          and table_name = '%(table_name)s'
                                      ) as a
                                where num=1 ''' % vars()
                out, err, code = cmdutils.exec_spark_sql(sql_value)
                out = out.split('\n')
                out.remove('')  # 去除最后的空字符
                dim_values = out
            elif isinstance(dim_value_cfg, list):
                dim_values = "','".join(map(str, dim_value_cfg))
                dim_values = "('" + dim_values + "')"
            elif not isinstance(dim_value_cfg, list):
                print("yaml配置错误，dim_value只能配置成list")
                sys.exit(1)

            sql_conditions = []
            unique_partitions = []

            sql_conditions.append("env = '%(env)s'" % vars())
            sql_conditions.append("group_id = %(monitor_id)s" % vars())
            sql_conditions.append("col_name = '%(col_name)s'" % vars())
            sql_conditions.append("cal_type = %(cal_type)s" % vars())
            unique_partitions.append("date")
            unique_partitions.append('env')
            unique_partitions.append('group_id')
            unique_partitions.append('col_name')
            unique_partitions.append('cal_type')
            if dim_value_cfg != [None]:  # 对应~
                unique_partitions.append('dim_value')

            if dim_value_cfg != [None]:  # 对应~
                sql_conditions.append("dim_value in %(dim_values)s" % vars())
            if 'cal_algo_json_conditions' in cfg:
                conditions = cfg.get('cal_algo_json_conditions')
                for condition in conditions:
                    condition_k = condition['k']
                    condition_v = str(condition['v']).replace("'", "\\'")
                    sql_conditions.append("get_json_object(cal_algo, '$.%(condition_k)s')='%(condition_v)s'" % vars())
                    unique_partitions.append("get_json_object(cal_algo, '$.%(condition_k)s')" % vars())
            else:
                conditions = []

            sql_where = " and ".join(sql_conditions)
            sql_window_partition = ", ".join(unique_partitions)

            # 历史-before_val, 当天-today_val
            sql = """select dim_value 
                           ,%(monitor_algo)s(case when date ='%(arg_date)s' then val else null end) as today_val  
                     from(
                          select *, row_number() over (partition by %(sql_window_partition)s order by cal_time desc) as rn
                          from  %(table_data_quality)s
                          where date ='%(arg_date)s' and %(sql_where)s
                          )
                          where rn = 1
                     group by dim_value
                  """ % vars()

            # prophet
            sql2 = """select date as ds
                            ,val as y
                      from(
                           select *, row_number() over (partition by %(sql_window_partition)s order by cal_time desc) as rn
                           from  %(table_data_quality)s
                           where %(sql_date_contitions)s and %(sql_where)s
                           ) as a
                      where rn = 1
                   """ % vars()

            sql_metas.append((sql, threshold_warning, threshold_error, col_name, cal_type, monitor_type, conditions,
                                  monitor_name, sql2, monitor_id))
        return sql_metas

    def __do_monitor_metrics(self, cfg):  # 比较历史
        target_table = self.__monitor_table
        target_date = self.__arg_date

        res_date = []
        res_table = []
        res_field = []
        res_target = []
        res_group = []
        res_before_val = []
        res_today_val = []
        res_volatility = []
        res_threshold = []
        res_abnormal = []

        for sql_meta in self.get_monitor_sqls(cfg):  # 循环cols
            sql = sql_meta[0]
            threshold_warning = str(sql_meta[1])
            threshold_error = str(sql_meta[2])
            col_name = sql_meta[3]
            cal_type = sql_meta[4]
            monitor_type = sql_meta[5]
            conditions = sql_meta[6]
            monitor_name = sql_meta[7]
            sql2 = sql_meta[8]
            monitor_id = sql_meta[9]
            unit = cfg.get('unit')

            file_path = os.path.abspath(__file__)
            current_path = os.path.abspath(os.path.dirname(file_path) + os.path.sep + ".")

            out, err, code = cmdutils.exec_spark_sql(sql)
            hive = "hive"
            excel_name = "warehouse_" + str(monitor_id) + "_" + col_name
            file_path = "%(current_path)s/%(excel_name)s.csv" % vars()
            cmd = """%(hive)s -e "
                     set mapreduce.job.queuename=warehouse;
                     set mapreduce.job.name=Hive:[warehouse-quality][monitor_prophet];
                     set hive.cli.print.header=true;
                     %(sql2)s" > %(file_path)s """ % vars()
            out2, err2, code2 = cmdutils.exec_shell(cmd)

            if str(code) == '0' and str(code2) == '0':
                df = pd.read_csv(file_path, sep="\t")
                m = Prophet()
                m.fit(df)
                future = m.make_future_dataframe(periods=7)
                forecast = m.predict(future)
                out2 = forecast[['ds', 'yhat']]
                print("out2:", out2)
                # 预测的当天值
                before_val = out2[out2['ds'] == target_date]['yhat']
                if str(before_val) == "NULL" or str(before_val) == "":
                    before_val = ""
                else:
                    before_val = float(before_val)

                outs = out.split('\n')
                outs.remove('')  # 去除最后的空字符
                for out in outs:
                    dim_value = out.split('\t')[0]
                    # 当天值
                    today_val = out.split("\t")[1]
                    if str(today_val) == "NULL" or str(today_val) == "":
                        today_val = ""
                    else:
                        today_val = float(today_val)
                    exceeded_threshold = "Normal"
                    volatility = ""  # 波动
                    monitor_threshold = threshold_warning
                    if str(before_val) == "":  # 历史为空
                        volatility = "First_run"
                        before_val = "No data"
                        if str(today_val) == "":  # 历史为空，当天为空
                            today_val = "No data"
                            exceeded_threshold = "Warning"
                        elif round(today_val * 10000, 2) == 0.00:  # 历史为空，当天为0
                            exceeded_threshold = "Warning"
                        else:  # 历史为空，当天非0
                            volatility = str(before_val) + " -> " + str(round(today_val / unit, 4))
                            exceeded_threshold = "Warning"
                            today_val = round(today_val / unit, 4)
                    elif round(before_val * 10000, 2) == 0.00:  # 历史为0
                        if str(today_val) == "":  # 历史为0，当天为空
                            today_val = "No data"
                            volatility = "No data today"
                            exceeded_threshold = "Error"
                        elif round(today_val * 10000, 2) == 0.00:  # 历史为0，当天为0
                            volatility = 0
                        else:  # 历史为0，当天非0
                            volatility = str(before_val) + " -> " + str(round(today_val / unit, 4))
                            exceeded_threshold = "Warning"
                            today_val = round(today_val / unit, 4)
                    elif round(before_val * 10000, 2) != 0.00:  # 历史非0
                        if str(today_val) == "":  # 历史非0，当天为空
                            today_val = "No data"
                            volatility = "No data today"
                            exceeded_threshold = "Error"
                            before_val = round(before_val / unit, 4)
                        elif round(today_val * 10000, 2) == 0.00:  # 历史非0，当天为0
                            volatility = 1
                            exceeded_threshold = "Error"
                            before_val = round(before_val / unit, 4)
                        else:  # 历史非0，当天非0
                            volatility = round((today_val - before_val) / before_val, 4)
                            before_val = round(before_val / unit, 4)
                            today_val = round(today_val / unit, 4)
                            if abs(volatility) >= float(threshold_error):
                                exceeded_threshold = "Error"
                                monitor_threshold = threshold_error
                            elif abs(volatility) >= float(threshold_warning):
                                exceeded_threshold = "Warning"
                                monitor_threshold = threshold_warning

                    if not (monitor_type == 3 and (exceeded_threshold == "Normal" or today_val == "No data")):  # 只保留该类型监控的异常数据
                        res_date.append(target_date)
                        res_table.append(target_table)
                        if conditions == []:
                            Monitor = monitor_name
                        else:
                            Monitor = conditions[0]["v"]
                        res_target.append(Monitor)
                        res_before_val.append(str(before_val))
                        res_today_val.append(str(today_val))
                        if isinstance(volatility, float):
                            volatility = str(volatility * 100) + "%"
                            res_volatility.append(volatility)
                        else:  # No Data
                            res_volatility.append(volatility)
                        monitor_threshold = str(float(monitor_threshold) * 100) + "%"
                        res_threshold.append(monitor_threshold)
                        res_abnormal.append(exceeded_threshold)
                        if cal_type == 1:
                            res_field.append(col_name)
                            res_group.append(u"否")
                        elif cal_type == 2:
                            target_field = conditions[1]["v"]
                            res_field.append(target_field)
                            res_group.append(col_name + ":" + str(dim_value))
                        elif cal_type == 3 and conditions == []:
                            res_field.append('')
                            res_group.append(col_name + ":" + str(dim_value))

        result = [res_date, res_table, res_field, res_target, res_group, res_before_val,
                  res_today_val, res_volatility, res_threshold, res_abnormal]
        row = len(res_date)
        return result, row

    def __do_sendmail(self, result, row_all, cfg):  # 比较历史模式发送邮件
        target_date = self.__arg_date
        target_table = self.__monitor_table
        warn_email = self.__warn_email
        email_string = ','.join(warn_email)
        note = cfg.get('note')
        if note is None:
            note = ""
        subject_cfg = cfg.get('subject')
        send_sms = cfg.get('send_sms')
        monitor_threshold = result[9]
        if "Error" in monitor_threshold:
            alert_level = "[Error]"
        elif "Warning" in monitor_threshold:
            alert_level = "[Warning]"
        else:
            alert_level = "[Normal]"
        if subject_cfg is None:
            subject_cfg = ''

        lenList = row_all
        title = [u'日期', u'监控表', u'监控字段', u'监控指标', u'是否分组', u'历史值', u'当日值', u'波动', u'已设界值', u'是否异常']
        monitor_results_html = convertToHtml(result, title, lenList)
        if self.__db == "test":
            staging_env = "[Staging] "
        else:
            staging_env = " "

        mail_content = u'\n'.join(
            [u"""<!DOCTYPE html>
                    <html>
                        <head>
                            <meta charset="UTF-8">
                            <title>data_quality_warning</title>
                            <style type="text/css">
                                table, th, td {
                                    border: 1px solid black;
                                    text-align: right;
                                    border-collapse: collapse;
                                }
                            </style>
                        </head>
                        <body>
                        """,
             note,

             u"<h2>监控结果</h2>",
             monitor_results_html,
             u"<br /><br />",
             u"</body></html>",
             ]
        )
        if warn_email is None or len(warn_email) == 0:
            warn_email = get_mail_list('etl-reports/data-quality')
        print("============================monitor==========================================")
        mail_subject = alert_level + "[warehouse_quality_prophet]" + staging_env + target_table + " " + subject_cfg + " " + target_date
        send_mail(to=warn_email, subject=mail_subject, content=mail_content, formation='html')
        print("[INFO] alert mail send success")

        if alert_level == "[Error]" and send_sms == 'Y':
            subprocess.call(
            """source /opt/mv-bash/helper-func.sh && send_alert_sms %(email_string)s warehouse-quality %(subject_cfg)s """ % vars(),
            shell=True)

    def __do_monitor_mail(self, cfg):  # 不比较变化趋势，直接作为异常数据发送邮件
        target_table = self.__monitor_table
        arg_date = self.__arg_date
        table_data_quality = self.__table_data_quality
        warn_email = self.__warn_email

        res_date = []
        res_table = []
        res_target = []
        res_group = []
        res_today_val = []
        res_today_val2 = []

        subject_cfg = cfg.get('subject')
        monitor_id = cfg.get('monitor_id')
        monitor_name = cfg.get('monitor_name')
        note = cfg.get('note')

        sql = """select dim_value, val, col_name
                   from(
                        select *, row_number() over (partition by group_id, dim_value order by cal_time desc) as rn
                        from  %(table_data_quality)s
                        where date = '%(arg_date)s' and group_id = %(monitor_id)s
                       )
                   where rn = 1
                   order by val, int(col_name) desc """ % vars()
        out1, err1, code1 = cmdutils.exec_spark_sql(sql)
        if str(code1) == '0':
            sql_result = out1.split("\n")
            if len(sql_result) - 1 < 10:
                n = len(sql_result) - 1  # 经split分割会多出一行
            else:
                n = 10
            for i in range(0, n):
                dim_value = sql_result[i].split("\t")[0]
                today_val = sql_result[i].split("\t")[1]
                today_val = float(today_val)
                today_val2 = sql_result[i].split("\t")[2]

                res_date.append(arg_date)
                res_table.append(target_table)
                res_target.append(monitor_name)
                res_group.append(dim_value)
                res_today_val.append(today_val)
                res_today_val2.append(today_val2)

        result = [res_date, res_table, res_target, res_group, res_today_val, res_today_val2]
        lenList = len(res_date)
        title = [u'日期', u'监控表', u'监控指标', u'分组', u'当日值', u'分母']
        monitor_results_html = convertToHtml_simple(result, title, lenList)
        mail_content = u'\n'.join(
            [u"""<!DOCTYPE html>
                            <html>
                                <head>
                                    <meta charset="UTF-8">
                                    <title>data_quality_warning</title>
                                    <style type="text/css">
                                        table, th, td {
                                            border: 1px solid black;
                                            text-align: right;
                                            border-collapse: collapse;
                                        }
                                    </style>
                                </head>
                                <body>
                                """,
             note,

             u"<h2>监控结果</h2>",
             monitor_results_html,
             u"<br /><br />",
             u"</body></html>",
             ]
        )
        if warn_email is None or len(warn_email) == 0:
            warn_email = get_mail_list('etl-reports/data-quality')
        print("============================monitor==========================================")
        send_mail(to=warn_email,
                  subject="[Warning]" + "[warehouse_quality] " + target_table + " " + subject_cfg + " " + arg_date,
                  content=mail_content, formation='html')
        print("[INFO] alert mail send success")


def convertToHtml(result, title, lenList):
    d = {}
    index = 0
    for t in title:
        d[t] = result[index]
        index = index + 1
    df = pd.DataFrame(d, index=[i + 1 for i in range(lenList)])
    df = df[title]
    html = df.style.applymap(lambda x: 'color: red' if x in ('Warning', 'Error') else 'color: green', subset=[u'是否异常']).render()
    return html


def convertToHtml_simple(result, title, lenList):  # 简化版邮件文本模板，发送内容均为异常数据
    d = {}
    index = 0
    for t in title:
        d[t] = result[index]
        index = index + 1
    df = pd.DataFrame(d, index=[i + 1 for i in range(lenList)])
    df = df[title]
    html = df.to_html(index=False, float_format=u'{:.4f}'.format)
    return html


if __name__ == '__main__':
    # 解析参数
    opts, args = getopt.gnu_getopt(sys.argv[1:], "d:e:b:t:", ["date=", "env=", "db=", "table="])
    for op, value in opts:
        if op in ['-d', '--date']:
            ARG_TODAY = value
        if op in ['-e', '--env']:
            ENV = value
        if op in ['-b', '--db']:
            DB = value
        if op in ['-t', '--table']:
            TABLE = value

    # noinspection PyUnboundLocalVariable
    config_in = read_config.read_config(ENV, TABLE, DB)
    table_name_in = config_in[0]
    db_name_in = config_in[1]
    env_in = config_in[2]
    warn_email_in = config_in[5]
    custom_in = config_in[6]

    monitor = Monitor(env_in, ARG_TODAY, table_name_in, db_name_in, warn_email_in, custom_in)
    monitor.do_execute()

    print('monitor success')
